package cn.com.agree.bigdata.vmdatahandle.service;

import cn.com.agree.bigdata.vmdatahandle.config.AppConfig;
import kong.unirest.HttpResponse;
import kong.unirest.Unirest;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.io.BufferedReader;
import java.io.StringReader;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

/**
 * @description:
 * @author: LiYuan
 * @date: 2024/7/9 15:54
 */

@Slf4j
public class WriteWmWorker implements Runnable {
    private static final String ClOSE_BRACE = "}";
    private static final String OPEN_BRACE = "{";
    private static final int STRING_BUILDER_INIT_SIZE = 2048;
    private static final String SYMBOL_SPACE = " ";
    private static final int METRIC_NAME_COPY_SIZE = 200;
    private static final int HOST_ID_SIZE = 25;
    private AppConfig appConfig;
    private ExecutorService executorService;

    public WriteWmWorker(AppConfig appConfig, ExecutorService executorService) {
        this.appConfig = appConfig;
        this.executorService = executorService;
    }

    @Override
    public void run() {
        log.info("begin read nodeexporter and write to vm");
        long startTime = System.currentTimeMillis();
        log.info("获取指标开始时间:{}", startTime);
        String url = getNodeExporterUrl();
        HttpResponse<String> response;
        try {
            response = Unirest.get(url).asString();
        } catch (Exception e) {
            log.error("获取node exporter指标异常:", e);
            return;
        }
        long receiveTime = System.currentTimeMillis();
        log.info("请求完成时间:{}，耗时:{}秒", receiveTime, (receiveTime-startTime)/1000.0);
        String metrics = response.getBody();
        long endTime = System.currentTimeMillis();
        log.info("获取指标结束时间:{}，耗时:{}秒", endTime, (endTime-startTime)/1000.0);

        // 把指标发送到vm
        int threads = appConfig.getVm().getThreads();
        if (threads > 0) {
            concurrentSendMetricsToWm(metrics, METRIC_NAME_COPY_SIZE, HOST_ID_SIZE);
        } else {
            sendMetricsToWm(metrics, METRIC_NAME_COPY_SIZE, HOST_ID_SIZE);
        }
    }

    public String getNodeExporterUrl() {
        String server = appConfig.getNodeExporter().getServer();
        return String.format("http://%s/metrics", server);
    }


    public String getVmUrl() {
        String server = appConfig.getVm().getServer();
        return String.format("http://%s/api/v1/import/prometheus", server);
    }

    public void sendMetricsToWm(String metrics, int metricNameSize, int hostIdSize) {
        int num = metricNameSize * hostIdSize;
        String url = getVmUrl();
        StringBuilder[] sbArr = new StringBuilder[num];
        for (int i = 0; i < num; i++) {
            sbArr[i] = new StringBuilder(STRING_BUILDER_INIT_SIZE);
        }
        long startTime = System.currentTimeMillis();
        log.info("处理数据开始时间:{}", startTime);

        try (BufferedReader reader = new BufferedReader(new StringReader(metrics))) {
            String line;
            while ((line = reader.readLine()) != null) {
                if (line.startsWith("#")) {
                    continue;
                }
                if (StringUtils.isBlank(line)) {
                    continue;
                }
                String[] newLineArr = batchParseMetricsLine(line, metricNameSize, hostIdSize);
                if (newLineArr == null) {
                    continue;
                }
                for (int i = 0; i < num; i++) {
                    sbArr[i].append(newLineArr[i]).append("\n");
                }
            }
        } catch (Exception e) {
            log.error("Exception:", e);
        }
        long handleEndTime = System.currentTimeMillis();
        log.info("处理数据结束时间:{},处理耗时:{}秒", handleEndTime, (handleEndTime-startTime)/1000.0);
        // 一次性发送全部指标
        int sendNum = 1;
        for (StringBuilder sb : sbArr) {
            long beginSendTime = System.currentTimeMillis();
            try {
                int length = sb.length();
                if (length > 1) {
                    // 最后的换行符不要发送
                    Unirest.post(url).body(sb.substring(0, length - 1)).asString();
                }
            } catch (Exception e) {
                log.error("发送指标给vm异常:", e);
            } finally {
                long endSendTime = System.currentTimeMillis();
                log.info("发送第{}笔指标结束时间:{},耗时:{}秒", sendNum, endSendTime, (endSendTime - beginSendTime) / 1000.0);
                sendNum++;
            }
        }
        long sendEndTime = System.currentTimeMillis();
        log.info("发送结束时间:{}，发送耗时:{}秒", sendEndTime, (sendEndTime-handleEndTime)/1000.0);
    }


    public void concurrentSendMetricsToWm(String metrics, int metricNameSize, int hostIdSize) {
        String url = getVmUrl();
        int num = metricNameSize * hostIdSize;
        StringBuilder[] sbArr = new StringBuilder[num];
        for (int i = 0; i < num; i++) {
            sbArr[i] = new StringBuilder(STRING_BUILDER_INIT_SIZE);
        }
        long startTime = System.currentTimeMillis();
        log.info("处理数据开始时间:{}", startTime);

        try (BufferedReader reader = new BufferedReader(new StringReader(metrics))) {
            String line;
            while ((line = reader.readLine()) != null) {
                if (line.startsWith("#")) {
                    continue;
                }
                if (StringUtils.isBlank(line)) {
                    continue;
                }
                String[] newLineArr = batchParseMetricsLine(line, metricNameSize, hostIdSize);
                if (newLineArr == null) {
                    continue;
                }
                for (int i = 0; i < num; i++) {
                    sbArr[i].append(newLineArr[i]).append("\n");
                }
            }
        } catch (Exception e) {
            log.error("Exception:", e);
        }
        long handleEndTime = System.currentTimeMillis();
        log.info("处理数据结束时间:{},处理耗时:{}秒", handleEndTime, (handleEndTime-startTime)/1000.0);
        // 一次性发送全部指标
        // 多线程发送
        final CountDownLatch finishLatch = new CountDownLatch(num);
        for (StringBuilder sb : sbArr) {
            executorService.submit(
                () -> {
                    long threadStartTime = System.currentTimeMillis();
                    try {
                        int length = sb.length();
                        if (length > 1) {
                            // 最后的换行符不要发送
                            Unirest.post(url).body(sb.substring(0, length - 1)).asString();
                            log.info("发送的指标为:{}", sb.substring(0, length - 1));
                        }
                    } catch (Exception e) {
                        log.error("发送指标给vm异常:", e);
                    } finally {
                        long threadEndTime = System.currentTimeMillis();
                        log.info("单条发送结束时间:{}，耗时:{}秒", threadEndTime,(threadEndTime - threadStartTime)/1000.0);
                        finishLatch.countDown();
                    }
                }
            );
        }

        // 全部线程发送完毕才进行下一次定时器发送
        try {
            finishLatch.await();
        } catch (InterruptedException e) {
            log.error("CountDownLatch await exception:", e);
        }
        long sendEndTime = System.currentTimeMillis();
        log.info("发送结束时间:{}，发送耗时:{}秒", sendEndTime, (sendEndTime-handleEndTime)/1000.0);
    }


    /**
     * 解析指标的每一行，
     * 如果有标签，则追加host_id标签；
     * 如果原来没标签，则增加{host_id: "XX"}
     * @param line
     * @param id
     * @return
     */
    public String parseMetricsLine(String line, int id) {
        int spaceIndex = StringUtils.lastIndexOf(line, SYMBOL_SPACE);
        if (spaceIndex == -1) {
            log.error("parse metrics line error:{}", line);
            return null;
        }
        String[] strArr = new String[2];
        strArr[0] = StringUtils.substring(line, 0, spaceIndex);
        strArr[1] = StringUtils.substring(line, spaceIndex + 1);

        int index = StringUtils.lastIndexOf(strArr[0], ClOSE_BRACE);
        if (index == -1) {
            return strArr[0] + "{host_id=\"" + id + "\"} " + strArr[1];
        }
        return StringUtils.substring(line,0, index) + ",host_id=\"" + id + "\"} " + strArr[1];
    }

    public String[] batchParseMetricsLine(String line, int metricsNameSize, int hostIdSize) {
        String[] newStrLineArr = new String[metricsNameSize * hostIdSize];
        int spaceIndex = StringUtils.lastIndexOf(line, SYMBOL_SPACE);
        if (spaceIndex == -1) {
            log.error("parse metrics line error:{}", line);
            return null;
        }
        String[] strArr = new String[2];
        strArr[0] = StringUtils.substring(line, 0, spaceIndex);
        strArr[1] = StringUtils.substring(line, spaceIndex + 1);

        int closeBraceIndex = StringUtils.lastIndexOf(strArr[0], ClOSE_BRACE);
        int openBraceIndex = StringUtils.indexOf(strArr[0], OPEN_BRACE);

        int lineIndex = 0;
        for (int i = 0; i < metricsNameSize; i++) {
            for(int j = 0; j < hostIdSize; j++) {
                // 没有标签的指标
                if (openBraceIndex == -1) {
                    // 有右括号则数据格式不对
                    if (closeBraceIndex != -1) {
                        log.error("parse metrics line error:{}", line);
                        return null;
                    }
                    newStrLineArr[lineIndex] = strArr[0] + "_" + i + "{host_id=\"" + j + "\"} " + strArr[1];
                    lineIndex++;
                    continue;
                }
                // 有标签的指标，没有右括号格式不对
                if (closeBraceIndex == -1) {
                    log.error("parse metrics line error:{}", line);
                    return null;
                }
                newStrLineArr[lineIndex] = StringUtils.substring(line, 0, openBraceIndex) + "_" + i + StringUtils.substring(line, openBraceIndex, closeBraceIndex) + ",host_id=\"" + j + "\"} " + strArr[1];
                lineIndex++;
            }
        }
        return newStrLineArr;
    }
}
